package org.apm.data.check;

import javax.management.Query;

import org.apache.log4j.LogManager;
import org.apache.log4j.Logger;
import org.apm.data.pusher.CommonMethod;
import org.elasticsearch.action.bulk.BulkProcessor;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.search.ClearScrollRequestBuilder;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.client.Client;
import org.elasticsearch.common.unit.TimeValue;
import org.elasticsearch.index.query.QueryBuilder;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.search.SearchHit;

public abstract class DataReIndex {
    private static final Logger LOGGER = LogManager.getLogger(DataReIndex.class);
    private static final int SCROLL_TIME = 60000;
    private static final int SCROLL_SIZE = 1000;

    private Client client = CommonMethod.devClient();
    private BulkProcessor esBulkProcessor = CommonMethod.EsBulkProcessor.DEV.processor();

    protected DataReIndex() {}

    protected DataReIndex(Client client, BulkProcessor esBulkProcessor) {
        this.client = client;
        this.esBulkProcessor = esBulkProcessor;
    }

    /**
     * check the name of the index is correct.
     * 
     * @param index the index name.
     * @return correct or not.
     */
    protected abstract boolean checkIndex(final String index);

    /**
     * get the query.
     */
    protected QueryBuilder query() {
        return QueryBuilders.matchAllQuery();
    }

    /**
     * process the source of the elasticsearch doc before reindexing.
     * 
     * @param dataSource the source string.
     * @return the analyzed string in Json format.
     */
    protected abstract DataBean processData(final String dataSource);

    /**
     * copy data from index to newIndex.
     * 
     * @param index data source
     * @param newIndex data destination.
     */
    public void copyIndexData(final String index, final String newIndex) {

        if (checkIndex(index)) {
            SearchResponse scrollResp = client.prepareSearch(index).setScroll(new TimeValue(SCROLL_TIME))
                    .setSize(SCROLL_SIZE).setQuery(query()).execute().actionGet();

            ClearScrollRequestBuilder clearRequest = client.prepareClearScroll();
            while (true) {
                for (SearchHit hit : scrollResp.getHits()) {
                    DataBean databean = processData(hit.getSourceAsString());
                    if (databean.isFlag()) {
                        esBulkProcessor.add(
                                new IndexRequest(newIndex, hit.getType(), hit.getId()).source(databean.getSource()));
                    }
                }

                scrollResp = client.prepareSearchScroll(scrollResp.getScrollId()).setScroll(new TimeValue(SCROLL_TIME))
                        .execute().actionGet();

                // Break condition: No hits are returned
                if (scrollResp.getHits().getHits().length == 0) {
                    break;
                } else {
                    LOGGER.info("[Analyzer-After][" + index + "] " + scrollResp.getHits().getHits().length + ".");
                }
            }
            
            try {
                Thread.sleep(100000);
            } catch (InterruptedException exception) {
                // TODO Auto-generated catch block
                exception.printStackTrace();
            }

            // 清缓存
            clearRequest.addScrollId(scrollResp.getScrollId()).execute().actionGet().isSucceeded();
        } else {
            LOGGER.error("Please use the correct reindex class.");
        }
    }

    protected static class DataBean {
        private boolean flag = false;
        private String source = "";

        public DataBean() {
            // TODO Auto-generated constructor stub
        }

        public DataBean(final String source) {
            this.source = source;
        }

        public DataBean(final String source, final boolean flag) {
            this.flag = flag;
            this.source = source;
        }

        public boolean isFlag() {
            return flag;
        }

        public void setFlag(boolean flag) {
            this.flag = flag;
        }

        public String getSource() {
            return source;
        }

        public void setSource(String source) {
            this.source = source;
        }
    }
}
